package drds.data_migrate.vo;

import drds.common.Author;
import drds.data_migrate.Authors;
import org.apache.commons.lang3.Validate;


/**
 * 这里主要是业务层面的处理
 */
@Author(name = Authors.LI_YANG)
public final class Messages
{
    public static final String stage = "stage";
    public static final String record_speed = "recordSpeed";
    public static final String percentage = "percentage";

    public static final String read_succeed_records = "readSucceedRecords";
    public static final String read_failed_records = "readFailedRecords";
    public static final String write_received_records = "writeReceivedRecords";
    public static final String write_failed_records = "writeFailedRecords";

    public static final String total_read_records = "totalReadRecords";
    public static final String wait_writer_time = "waitWriterTime";
    public static final String wait_reader_time = "waitReaderTime";


    public static final String total_error_records = "totalErrorRecords";
    public static final String write_succeed_records = "writeSucceedRecords";


    public static State getReportCommunication(State now, State old, int totalStage)
    {
        Validate.isTrue(now != null && old != null,
                "为汇报准备的新旧metric不能为null");

        long totalReadRecords = getTotalReadRecords(now);

        now.setLong(total_read_records, totalReadRecords);

        now.setLong(total_error_records, getTotalErrorRecords(now));

        now.setLong(write_succeed_records, getWriteSucceedRecords(now));


        long timeInterval = now.getTimestamp() - old.getTimestamp();
        long sec = timeInterval <= 1000 ? 1 : timeInterval / 1000;

        long recordsSpeed = (totalReadRecords
                - getTotalReadRecords(old)) / sec;


        now.setLong(record_speed, recordsSpeed < 0 ? 0 : recordsSpeed);
        now.setDouble(percentage, now.getLong(stage) / (double) totalStage);

        if (old.getThrowable() != null)
        {
            now.setThrowable(old.getThrowable());
        }

        return now;
    }

    public static long getTotalReadRecords(final State state)
    {
        return state.getLong(read_succeed_records) +
                state.getLong(read_failed_records);
    }


    public static long getTotalErrorRecords(final State state)
    {
        return state.getLong(read_failed_records) +
                state.getLong(write_failed_records);
    }


    public static long getWriteSucceedRecords(final State state)
    {
        return state.getLong(write_received_records) -
                state.getLong(write_failed_records);
    }


}
